package FPGrowth

import java.io.File
import java.util

import org.apache.commons.io.FileUtils
import org.apache.spark.mllib.fpm.{FPGrowth, FPGrowthModel}
import org.apache.spark.mllib.recommendation.Rating
import org.apache.spark.{SparkConf, SparkContext}

object FPTreeDemo {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", new File("").getAbsolutePath)

    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("My App")
    val sc: SparkContext = new SparkContext(conf)

    val inputFile1: String = "topics/FPGrowthList.csv"

    val modelPath="FPGrowthModel"
    val model=if(!new File(modelPath).exists()){
      val train = sc.textFile(inputFile1).flatMap(_.split("\n"))//.filter(_ contains "|")
        .map(_.split("\\|")(1))
        .map(_.split(",").take(30))
      val model = new FPGrowth()
        .setMinSupport(0.0001)
        .setNumPartitions(8)
        .run(train)
      model.save(sc,"FPGrowthModel")
      model
    }else{
      FPGrowthModel.load(sc,modelPath)
    }
//    model.freqItemsets.filter(e=>{
//      e.items.length==2 && e.items.contains("19596107")
//    }).collect().sortBy(_.freq)
//      .foreach(e=>println(e.items.map(perProc.maps.get).mkString(",")+"|"+e.freq))

//    model.itemSupport.foreach(e=>println(e._1+","+perProc.maps.get(e._1)+","+e._2+"\n"))

//    val outfile = new File("1ans")
//    FileUtils.write(outfile, "", false)
//    val ans=new StringBuilder
//    model.freqItemsets.foreach(itemset=>{
//      val sb=new StringBuilder
//      itemset.javaItems.forEach(e=>sb++=","+e.toInt)
//      ans++=sb.substring(1) + "|" + itemset.freq+"\n"
//      if(ans.length>1e5){
//        FileUtils.write(outfile, ans.toString(), true)
//        ans.clear()
//      }
//    })
//    FileUtils.write(outfile, ans.toString(), true)


//    val ans2=new StringBuilder
//    val outfile2 = new File("1ans2")
//    FileUtils.write(outfile2, "", false)
//    model.generateAssociationRules(0.4).foreach(rule=>{
//      ans2++=rule.antecedent.mkString(",")+"|"+ rule.consequent.mkString(",")+"|"+ rule.confidence+"\n"
//      if(ans2.length>1e5){
//        FileUtils.write(outfile2, ans2.toString(), true)
//        ans2.clear()
//      }
//    })
//    FileUtils.write(outfile2, ans2.toString(), true)
  }

}
